package com.gper.edu.flink.stream;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.IOException;

/**
 * @Author: ellis.guan
 * @Description:
 * @Date: 2020/11/8 9:53
 * @Version: 1.0
 */
public abstract class StreamEnvironment {
    protected StreamExecutionEnvironment env;
    protected StreamTableEnvironment tableEnv;
    private static final int DEFAULT_CHECKPOINT_TIME = 5 * 60 * 1000;
    private static final int DEFAULT_CHECKPOINT_TIMEOUT = 10 * 60 * 1000;

    public Configuration getConfiguration(){
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        return configuration;
    }

    /**
     * 获取socketTextStream数据源
     * @return
     */
    public DataStreamSource<String> getDataStreamSource(){
        return env.socketTextStream("zeppelin.gxc.com", 9080);
    }

    public void setParallelism(int parallelism){
        env.setParallelism(parallelism);
    }
    /**
     * 初始化环境变量
     */
    public void initEnvironment() throws IOException {
        //使用本地环境 且带有Flink webUI
        env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(getConfiguration());
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //使用Blink planner
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        tableEnv = StreamTableEnvironment.create(env, settings);
        setCheckPointConfig();
//        setMiniBatch();
    }

    private void setMiniBatch() {
        tableEnv.getConfig().getConfiguration().setString("table.exec.mini-batch.enabled", "true");
        tableEnv.getConfig().getConfiguration().setString("table.exec.mini-batch.allow-latency", "10 s");
        tableEnv.getConfig().getConfiguration().setString("table.exec.mini-batch.size", "5000");
    }

    private void setCheckPointConfig() throws IOException {
        // start a checkpoint every 5 minute
        env.enableCheckpointing(DEFAULT_CHECKPOINT_TIME);
        //以下是设置checkpoint的高级选项:
        //1. set mode to exactly-once (this is the default)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //2. 设置checkpoint之间的最小时间间隔,单位ms
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(60000);
        //3. 设置checkpoint的超时时间，单位是ms
        env.getCheckpointConfig().setCheckpointTimeout(DEFAULT_CHECKPOINT_TIMEOUT);
        // allow only one checkpoint to be in progress at the same time
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // enable externalized checkpoints which are retained after job cancellation
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // allow job recovery fallback to checkpoint when there is a more recent savepoint
        env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
        // enables the experimental unaligned checkpoints
        env.getCheckpointConfig().enableUnalignedCheckpoints();

        String checkPointPath = "hdfs:///flink/checkpoint";
        if (System.getProperty("os.name").startsWith("Windows")) {
            checkPointPath = "file:///d:/flink/checkpoint";
        }
        RocksDBStateBackend rocksDB = new RocksDBStateBackend(checkPointPath, true);
        rocksDB.setNumberOfTransferThreads(6);
        env.setStateBackend(rocksDB);
    }

    public abstract void execute() throws Exception;

    public void runJob() {
        try {
            initEnvironment();
            execute();
            env.execute(this.getClass().getName());
        }catch (Exception e){
            throw new RuntimeException(e);
        }
    }

}
